package com.howe.gateway.balancer;

import com.alibaba.fastjson2.JSONArray;
import com.howe.gateway.entity.DubboEndpointsConfig;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.loadbalancer.core.*;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class CustomLoadBalancer implements ReactorServiceInstanceLoadBalancer {

    private static final Log log = LogFactory.getLog(CustomLoadBalancer.class);

    final AtomicInteger position;

    final String serviceId;

    ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;

    public CustomLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider,
                              String serviceId) {
        this(serviceInstanceListSupplierProvider, serviceId, new Random().nextInt(1000));
    }

    public CustomLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider,
                              String serviceId, int seedPosition) {
        this.serviceId = serviceId;
        this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;
        this.position = new AtomicInteger(seedPosition);
    }

    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider
                .getIfAvailable(NoopServiceInstanceListSupplier::new);
        return supplier.get(request).next()
                .map(serviceInstances -> processInstanceResponse(supplier, serviceInstances));
    }

    private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier,
                                                              List<ServiceInstance> serviceInstances) {
        Response<ServiceInstance> serviceInstanceResponse = getInstanceResponse(serviceInstances);
        if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {
            ((SelectedInstanceCallback) supplier).selectedServiceInstance(serviceInstanceResponse.getServer());
        }
        return serviceInstanceResponse;
    }

    private final static String DUBBO_ENDPOINTS = "dubbo.endpoints";

    private final static String REST = "rest";

    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances) {
        if (instances.isEmpty()) {
            if (log.isWarnEnabled()) {
                log.warn("No servers available for service: " + serviceId);
            }
            return new EmptyResponse();
        }
        //基于roundLoadBalance逻辑 增加dubbo示例的元数据解析
        // dubbo示例中的 port 优先级 > http port
        /*
        {dubbo.metadata-service.url-params=
        {"prefer.serialization":"fastjson2,hessian2","version":"1.0.0","dubbo":"2.0.2","release":"3.2.8","side":"provider","port":"20880","protocol":"dubbo"},
        nacos.instanceId=2.0.111.104#20880#null#provider,
        nacos.weight=1.0, nacos.cluster=DEFAULT,
        nacos.ephemeral=true, nacos.healthy=true,
        dubbo.endpoints=[{"port":80,"protocol":"rest"},{"port":20880,"protocol":"dubbo"}],
        dubbo.metadata.revision=ff1ac96b780b2e6330bb1dc551493b49, dubbo.metadata.storage-type=local, timestamp=1718087737081}
         */
        List<ServiceInstance> dubboInstances = instances.stream().filter(instance -> {
            Map<String, String> metadata = instance.getMetadata();
            String endpoints = metadata.get(DUBBO_ENDPOINTS);
            //如果元数据包含dubbo.endpoints属性 默认为dubbo应用
            if(Objects.nonNull(endpoints)){
                List<DubboEndpointsConfig> configs = JSONArray.parseArray(endpoints, DubboEndpointsConfig.class);
                //判断该dubbo应用是否开启rest端口
                return configs.stream().anyMatch(config -> config.getProtocol().equals(REST));
            }
            return false;
        }).toList();

        List<ServiceInstance> httpInstances = instances.stream()
                .filter(instance -> "SPRING_CLOUD".equals(instance.getMetadata().get("preserved.register.source")))
                .toList();

        // replace instances
        instances = dubboInstances.isEmpty() ? httpInstances : dubboInstances;

        if (instances.size() == 1) {
            return new DefaultResponse(instances.get(0));
        }

        int pos = this.position.incrementAndGet() & Integer.MAX_VALUE;

        ServiceInstance instance = instances.get(pos % instances.size());

        return new DefaultResponse(instance);
    }

}
